package com.hadoop.tuning;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.SnappyCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;


/*
 * Snappy는 압축률보다는 빠르게 압축을 수행하는 것을 목표로 구글에서 개발
 * # install
 * 1. tar xvfz snappy-x.x.tar.gz
 * 2. .configure/ --enable-shared (make 파일 생성)
 * 3. make (소스코드 Compile 및 바이너리 파일 생성)
 * 4. make install (snappy에서 지정한 디렉터리로 옮김)
 * 5. Snappy 라이브러리를 하둡의 네이트브 라이브러리 디렉터리로 복사
 * cp /usr/local/lib/libsnappy.*  /home/hadoop/hadoop/lib/native/Linux-i386-32/
 *
 */
public class ArrivalDelayCountWithSnappy {
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		
		//Map 출력 압축 설정
		conf.setBoolean("mapred.compress.map.output", true);
		conf.set("mapred.map.output.compression.codec",
				"org.apache.hadoop.io.compress.SnappyCodec");

		if (args.length != 2) {
			System.err
					.println("Usage: ArrivalDelayCountWithSnappy <input> <output>");
			System.exit(2);
		}
		
		Job job = new Job(conf, "ArrivalDelayCountWithSnappy");

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		job.setJarByClass(ArrivalDelayCountWithSnappy.class);
		job.setMapperClass(ArrivalDelayCountMapper.class);
		job.setReducerClass(DelayCountReducer.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);

		SequenceFileOutputFormat.setCompressOutput(job, true);
		SequenceFileOutputFormat.setOutputCompressorClass(job,
				SnappyCodec.class);
		SequenceFileOutputFormat.setOutputCompressionType(job,
				CompressionType.BLOCK);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		job.waitForCompletion(true);
	}
}
